package com.easydev.mq.rabbit.pool;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.easydev.mq.exception.MessageException;
import com.easydev.mq.rabbit.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * <Description> rabbit connection类，一个connection包含多个channel<br>
 * 用于管理connection及及开启的虚拟线程<br>
 * 
 * @author lx<br>
 * @version 1.0<br>
 * @CreateDate 2016年3月11日 <br>
 */
public class RabbitConnection {

    private final static Logger LOGGER = LoggerFactory.getLogger(RabbitConnection.class);

    /*
     * rabbitMq tcp链接
     */
    private Connection connection;

    /*
     * 当前connection中channel
     */
    private List<Channel> channels;

    public RabbitConnection(String virtualHost) throws MessageException {
        LOGGER.debug("===== create connection =====");
        try {
            this.connection = createConnection(virtualHost);
            this.channels = new ArrayList<Channel>();
        } catch (Exception e) {
            LOGGER.error("create connection failed!~please check " + RabbitMQConfig.CONFIG_FILE);
            throw new MessageException("获取connection异常", e);
        }
    }

    /**
     * Description: 创建一个channel，并将解决存放在channels中<br>
     * 
     * @author lx<br>
     * @return <br>
     * @throws MessageException
     */
    public Channel createChannel() throws MessageException {
        Channel channel = null;
        try {
            channel = connection.createChannel();
            this.channels.add(channel);
        } catch (IOException e) {
            LOGGER.error("create channel failed", e);
            throw new MessageException("create channel failed", e);
        }
        return channel;
    }

    /**
     * Description: 获得当前connection中channel数量<br>
     * 
     * @author lx<br>
     * @return int<br>
     */
    public int size() {
        int num = 0;
        if (null != channels) {
            num = channels.size();
        }
        return num;
    }

    /**
     * Description: 删除当前channels中的对象，不等同于删除rabbit connection中的channel<br>
     * 
     * @author lx<br>
     * @param channel <br>
     */
    public void remove(Channel channel) {
        channels.remove(channel);
    }

    /**
     * Description: 判断CONNECTION是否有效<br>
     * 
     * @author lx<br>
     * @return 如果连接保持活动 则返回TRUE<br>
     */
    public boolean valid() {
        boolean flag = false;
        if (null != connection && connection.isOpen()) {
            flag = true;
        }
        return flag;
    }

    /**
     * Description: 获得当前CONNECTION下所有CHANNEL<br>
     * 
     * @return 当前CONNECTION下所有CHANNEL<br>
     */
    public List<Channel> getSubChannels() {
        return channels;
    }

    /**
     * Description: 创建一个与rabbit服务器的物理连接 <br>
     * 
     * @author lx <br>
     * @return com.rabbitmq.client.Connection
     * @throws IOException
     * @throws TimeoutException <br>
     */
    private Connection createConnection(String virtualHost) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitMQConfig.HOST);
        factory.setPort(Integer.parseInt(RabbitMQConfig.PORT));
        factory.setUsername(RabbitMQConfig.USERNAME);
        factory.setPassword(RabbitMQConfig.PASSWORD);
        if (StringUtils.isNotBlank(virtualHost)) {
            factory.setVirtualHost(virtualHost);
        }
        if (StringUtils.isNotBlank(RabbitMQConfig.CONNECTIONTIMEOUT)) {
            factory.setConnectionTimeout(Integer.parseInt(RabbitMQConfig.CONNECTIONTIMEOUT));
        }
        // 判断是否消费者开启线程池
        if (StringUtils.isNotBlank(RabbitMQConfig.THREAD_POOL)) {
            ExecutorService es = Executors.newFixedThreadPool(Integer.parseInt(RabbitMQConfig.THREAD_POOL));
            return factory.newConnection(es);
        }
        return factory.newConnection();
    }

    public void colse() {
        try {
            connection.close();
        } catch (IOException e) {
        }
    }
}
